package com.demo.rocketmq.config;

import com.demo.rocketmq.constants.DelayLevelEnum;
import com.demo.rocketmq.constants.RocketMqConstants;
import com.demo.rocketmq.producer.DelayProducer;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Configuration;
import org.springframework.lang.NonNull;

import java.util.List;
import java.util.Map;

/**
 * 项目启动后创建consumer，专门消费proxy-topic延时消息和重新再投不同延时level消息
 * @author K
 */
@Slf4j
@Configuration
public class RocketMQDelayConfig implements SmartInitializingSingleton, ApplicationContextAware {

    private ApplicationContext applicationContext;

    @Override
    public void setApplicationContext(@NonNull ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    @SneakyThrows
    @Override
    public void afterSingletonsInstantiated() {
        log.info("RocketMQConfig 初始化消费端");
        RocketMQTemplate template = applicationContext.getBean(RocketMQTemplate.class);
        DefaultMQProducer defaultMqProducer = template.getProducer();

        DelayProducer delayProducer = applicationContext.getBean(DelayProducer.class);

        DefaultMQPushConsumer defaultMqPushConsumer = new DefaultMQPushConsumer(defaultMqProducer.getProducerGroup());
        // 绑定NameServer
        defaultMqPushConsumer.setNamesrvAddr(defaultMqProducer.getNamesrvAddr());
        // 绑定消费者
        defaultMqPushConsumer.setMessageListener(new DelayMqMessageConsumer(delayProducer));
        defaultMqPushConsumer.subscribe(RocketMqConstants.PROXY_TOPIC, "*");
        defaultMqPushConsumer.start();
    }

    private static class DelayMqMessageConsumer implements MessageListenerConcurrently {

        private final DelayProducer delayProducer;

        public DelayMqMessageConsumer(DelayProducer delayProducer) {
            this.delayProducer = delayProducer;
        }

        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msg, ConsumeConcurrentlyContext context) {

            try {
                for (MessageExt messageExt : msg) {
                    Map<String, String> properties = messageExt.getProperties();

                    String topic = properties.get(RocketMqConstants.ORIGINAL_TOPIC);
                    String times = properties.get(RocketMqConstants.TIMES);
                    String tag = properties.get(RocketMqConstants.ORIGINAL_TAG);
                    String keys = properties.get(RocketMqConstants.ORIGINAL_KEY);
                    String uuid = properties.get(RocketMqConstants.ORIGINAL_UUID);
                    if (StringUtils.isBlank(topic)) {
                        log.error("消息id：{}，消息uuid：{}，未收到原topic， tag：{}，属性：{}", messageExt.getMsgId(), uuid, tag,
                                properties);
                        continue;
                    }
                    if (StringUtils.isBlank(times)) {
                        int delayTimeLevel = messageExt.getDelayTimeLevel();
                        times = String.valueOf(DelayLevelEnum.getTime(delayTimeLevel));
                        log.error("消息id：{}，消息uuId {}，未收到延时时间，消息topic：{}，tag：{}， 属性：{}。 重新设置延迟时间：{}",
                                messageExt.getMsgId(), uuid, topic, tag, properties, times);
                    }
                    properties.remove(RocketMqConstants.TIMES);
                    log.info("消息id：{}, 消息uuId：{}, --topic: {}-- tags: {} #####body:{}", messageExt.getMsgId(), uuid,
                            messageExt.getTopic(), messageExt.getTags(), new String(messageExt.getBody()));
                    Message message = new Message();
                    message.setTopic(topic);
                    if (StringUtils.isNotBlank(tag)) {
                        message.setTags(tag);
                    }
                    if (StringUtils.isNotBlank(keys)) {
                        message.setKeys(keys);
                    }
                    if (StringUtils.isNotBlank(uuid)) {
                        message.putUserProperty(RocketMqConstants.ORIGINAL_UUID, uuid);
                    }
                    message.setBody(messageExt.getBody());
                    delayProducer.sendDelay(message, Integer.parseInt(times));
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            } catch (Exception e) {
                log.error("消息发送失败", e);
            }
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    }
}
